package bma.common.netty.client;

import java.net.SocketAddress;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineException;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import bma.common.langutil.ai.stack.AIStack;
import bma.common.langutil.core.DateTimeUtil;
import bma.common.langutil.pipeline.CommonPipelineBuilder;
import bma.common.netty.ChannelPipelineFactory4Builder;

public class NettyClientBootstrap extends ClientBootstrap {

	protected Executor bossExecutor;
	protected Executor workerExecutor;
	protected int worker = 5;

	public NettyClientBootstrap() {
		super();
	}

	public NettyClientBootstrap(ChannelFactory channelFactory) {
		super(channelFactory);
	}

	public int getWorker() {
		return worker;
	}

	public void setWorker(int worker) {
		this.worker = worker;
	}

	public Executor getBossExecutor() {
		return bossExecutor;
	}

	public void setBossExecutor(Executor bossExecutor) {
		this.bossExecutor = bossExecutor;
	}

	public Executor getWorkerExecutor() {
		return workerExecutor;
	}

	public void setWorkerExecutor(Executor workerExecutor) {
		this.workerExecutor = workerExecutor;
	}

	public void setConnectTimeout(int connectTimeout) {
		if (connectTimeout > 0) {
			setOption("connectTimeoutMillis", connectTimeout);
		}
	}

	public void setConnectTimeoutValue(String s) {
		setConnectTimeout((int) DateTimeUtil.parsePeriodValue(s, 0));
	}

	public NettyClientBootstrap init() {
		try {
			this.getFactory();
			return this;
		} catch (IllegalStateException e) {
		}

		if (this.bossExecutor == null) {
			this.bossExecutor = Executors.newCachedThreadPool();
		}
		if (this.workerExecutor == null) {
			int ts = Runtime.getRuntime().availableProcessors() * 2;
			this.workerExecutor = new ThreadPoolExecutor(Math.min(ts, worker),
					worker, 0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>());
		}
		this.setFactory(new NioClientSocketChannelFactory(bossExecutor,
				workerExecutor, worker));
		return this;
	}

	@Override
	public ChannelFuture connect(SocketAddress remoteAddress,
			SocketAddress localAddress) {
		return connect(remoteAddress, localAddress, getPipelineFactory());
	}

	public ChannelFuture connect(SocketAddress remoteAddress,
			SocketAddress localAddress,
			CommonPipelineBuilder<ChannelHandler> builder) {
		return connect(remoteAddress, localAddress,
				new ChannelPipelineFactory4Builder(builder));
	}

	public ChannelFuture connect(SocketAddress remoteAddress,
			SocketAddress localAddress, ChannelPipelineFactory fac) {

		if (remoteAddress == null) {
			throw new NullPointerException("remoteAddress");
		}

		if (fac == null) {
			fac = getPipelineFactory();
		}
		ChannelPipeline pipeline;
		try {
			pipeline = fac == null ? Channels.pipeline() : fac.getPipeline();
		} catch (Exception e) {
			throw new ChannelPipelineException(
					"Failed to initialize a pipeline.", e);
		}

		// Set the options.
		Channel ch = getFactory().newChannel(pipeline);
		boolean success = false;
		try {
			ch.getConfig().setOptions(getOptions());
			success = true;
		} finally {
			if (!success) {
				ch.close();
			}
		}

		// Bind.
		if (localAddress != null) {
			ch.bind(localAddress);
		}

		// Connect.
		return ch.connect(remoteAddress);
	}

	public boolean connect(final AIStack<NettyClient> stack,
			SocketAddress remoteAddress, SocketAddress localAddress,
			ChannelPipelineFactory fac) {
		ChannelFuture cf = this.connect(remoteAddress, localAddress, fac);
		cf.addListener(new ChannelFutureListener() {

			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				if (future.isSuccess()) {
					Channel channel = future.getChannel();
					try {
						NettyClient nc = NettyClient.bind(channel);
						stack.success(nc);
					} catch (Exception e) {
						stack.failure(e);
					}
				} else {
					stack.failure(future.getCause());
				}
			}
		});
		return false;
	}
}
